package jrain.fw.mq.ons;

import java.util.Collection;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.OrderProducer;

import jrain.fw.core.business.BusinessConst;
import jrain.fw.core.cfg.CfgService;
import jrain.fw.core.log.Logger;
import jrain.fw.core.log.LoggerFactory;
import jrain.fw.core.mq.MqProducerService;
import jrain.fw.core.service.DestroyService;
import jrain.fw.core.service.InitService;
import jrain.fw.core.service.Service;
import jrain.fw.core.service.ServiceConst;
import jrain.fw.core.utils.FwUtils;
import jrain.utils.collection.CollectionUtils;
import jrain.utils.collection.RMap;
import jrain.utils.consts.BaseConst;
import jrain.utils.exception.ServiceException;
import jrain.utils.lang.LangUtils;
import jrain.utils.lang.StringUtils;

@Service(name = "ons", group = ServiceConst.S_MQ_PRODUCER_SERVICE, module = ServiceConst.M_MQ)
public class MqOnsProducerService implements MqProducerService, InitService, DestroyService {

	private static final AtomicBoolean init = new AtomicBoolean(false);
	private static final ConcurrentMap<String, Object> producerMap = new ConcurrentHashMap<>();
	private static final Logger logger = LoggerFactory.getLogger(MqOnsProducerService.class);

	public String send(String group, String topic, String tag, String data, String key) {
		Message message = new Message(topic, tag, data.getBytes(BaseConst.UTF8));
		if (key != null && key.length() != 0) {
			message.setKey(key);
		}
		Object obj = producerMap.get(group);

		if (obj != null) {
			if (obj instanceof Producer) {
				return ((Producer) obj).send(message).getMessageId();
			} else {
				String errorMsg = "producer are not invalid!,group=" + group;
				throw new ServiceException(errorMsg);
			}
		} else {
			String errorMsg = "producer are not initialized!,group=" + group;
			throw new ServiceException(errorMsg);
		}

	}

	public String sendOrder(String group, String topic, String tag, String data, String key) {
		return sendOrder(group, topic, tag, data, key, "global");
	}

	public String sendOrder(String group, String topic, String tag, String data, String key, String shardingKey) {
		Message message = new Message(topic, tag, data.getBytes(BaseConst.UTF8));
		if (key != null && key.length() != 0) {
			message.setKey(key);
		}
		Object obj = producerMap.get(group);
		if (obj != null) {
			if (obj instanceof OrderProducer) {
				return ((OrderProducer) obj).send(message, shardingKey).getMessageId();
			} else {
				String errorMsg = "producer are not invalid!,group=" + group;
				throw new ServiceException(errorMsg);
			}
		} else {
			String errorMsg = "producer are not initialized!,group=" + group;
			throw new ServiceException(errorMsg);
		}
	}

	public void init() {
		if (!init.getAndSet(true)) {
			CfgService cfgService = FwUtils.getCfgService();
			RMap<String, Object> cfgMap = cfgService.getCfgMap(ServiceConst.M_MQ, null);
			if (cfgMap != null) {
				List<String> groups =CollectionUtils.valueOfList(StringUtils.trimNull(cfgMap.get("mq.producer.gorups")));
				for (String group : groups) {
					boolean enable = LangUtils.parseBoolean(cfgMap.get("mq.producer."+group+".enable"), false);
					boolean order = LangUtils.parseBoolean(cfgMap.get("mq.producer."+group+".order"), false);
					String producerId = StringUtils.trimNull(cfgMap.get("mq.producer."+group+".id"));
					String accessKey = StringUtils.trimNull(cfgMap.get("mq.producer."+group+".access.key"));
					String secretKey = StringUtils.trimNull(cfgMap.get("mq.producer."+group+".secret.key"));
					String onsAddr = StringUtils.trimNull(cfgMap.get("mq.producer."+group+".ons.addr"));
					if(enable) {
						
						Properties properties = new Properties();
						properties.put(PropertyKeyConst.ProducerId, producerId);
						// AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
						properties.put(PropertyKeyConst.AccessKey, accessKey);
						// SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
						properties.put(PropertyKeyConst.SecretKey, secretKey);
						// 设置 TCP 接入域名（此处以公共云生产环境为例）
						properties.put(PropertyKeyConst.ONSAddr, onsAddr);
						if(order) {
							OrderProducer orderProducer = ONSFactory.createOrderProducer(properties);
							producerMap.putIfAbsent(group, orderProducer);
							logger.info(BusinessConst.LOG_BOOT_REQUEST,"{} producer order started!",group);
							orderProducer.start();
						}else {
							Producer producer = ONSFactory.createProducer(properties);
							producerMap.putIfAbsent(group, producer);
							logger.info(BusinessConst.LOG_BOOT_REQUEST,"{} producer started!",group);
							producer.start();
						}
						
					}
				}
				logger.info(BusinessConst.LOG_BOOT_REQUEST,"{} classpath started!",ServiceConst.S_MQ_PRODUCER_SERVICE);
			}
		}
	}

	@Override
	public void destroy() {
		if (init.get()) {
			Collection<Object> vlalues = producerMap.values();
			for (Object object : vlalues) {
				if (object != null) {
					if (object instanceof Producer) {
						try {
							((Producer) object).shutdown();
						} catch (Throwable t) {
							logger.error(BusinessConst.LOG_BOOT_REQUEST, t.getMessage(), t);
						}
					}
					if (object instanceof OrderProducer) {
						try {
							((OrderProducer) object).shutdown();
						} catch (Throwable t) {
							logger.error(BusinessConst.LOG_BOOT_REQUEST, t.getMessage(), t);
						}
					}
				}
			}
			logger.info(BusinessConst.LOG_BOOT_REQUEST,"{} classpath stoped!",ServiceConst.S_MQ_PRODUCER_SERVICE);
			init.set(false);
		}
	}

}
